package com.sunyur.tailor.channel.config.middle;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description kafka消息者配置类
 */
@Configuration
@ConditionalOnProperty(name = "spring.kafka.switch", havingValue = "open")
public class KafkaConsumerConfig {

    @Value("${spring.kafka.consumer.max-poll-records}")
    private Integer maxPollRecords;
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;
    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private Boolean enableAutoCommit;
    @Value("${spring.kafka.consumer.threads}")
    private Integer threads;
    @Value("${spring.kafka.consumer.session-timeout}")
    private Integer sessionTimeout;
    @Value("${spring.kafka.consumer.request-timeout}")
    private Integer requestTimeout;

    /**
     * OA
     */
    @Value("${spring.kafka.oa.bootstrap-servers}")
    private String oaServers;
    @Value("${spring.kafka.oa.consumer.group-id}")
    private String oaGroupId;
    /**
     * MDM
     */
    @Value("${spring.kafka.mdm.bootstrap-servers}")
    private String mdmServers;
    @Value("${spring.kafka.mdm.consumer.group-id}")
    private String mdmGroupId;

    private Map<String, Object> consumerConfigs(String groupId, String bootstrapServers) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeout);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, ConsumerConfig.DEFAULT_FETCH_MAX_BYTES);
        return props;
    }

    /**
     * 消费者ContainerFactory
     */
    private KafkaListenerContainerFactory consumerContainerFactory(Map<String, Object> configs){
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(configs));
        factory.setConcurrency(threads);
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

    /**
     * @Description: oa ContainerFactory
     */
    @Bean
    public KafkaListenerContainerFactory oaKafkaContainerFactory() {
        return consumerContainerFactory(consumerConfigs(oaGroupId, oaServers));
    }

    /**
     * @Description: mdm ContainerFactory
     */
    @Bean
    public KafkaListenerContainerFactory mdmKafkaContainerFactory() {
        return consumerContainerFactory(consumerConfigs(mdmGroupId, mdmServers));
    }
}
